from absl import app, flags
import datetime
import logging
import pandas as pd
import warnings
import os

from coin.base.timestamp import get_timestamp
from coin.tool.strat_monitor.handlers.check_balance_handler import BalanceChecker
from coin.tool.strat_monitor.util.display_util import (
    convert_balance_checker_summary_for_display,
    print_balance_checker_summary,
)
from coin.base.param_util import split_string
from coin.tool.strat_monitor.util.monitor_util import get_topic_from_strat
from coin.util.queue.config import KafkaConfig
from coin.util.queue.tools.kafka_archive import run_from_kafka_archive
from coin.support.pta.logic.pta_stats_calculator import get_accounting_currency
from coin.base.query_util import query_exchange_rates
from xunkemgmt_client.client.util.query_util import query_strategies
from xunkemgmt_client.tool.slack_noti import send_to_slack

FLAGS = flags.FLAGS

warnings.filterwarnings('ignore')

EXCLUDED_STRATS = []

EXCLUDED_MARKETS = [('Futures', 'Prex')]


def send_balance_checker_summary(display_df, slack_receiver):
  slack_df = display_df[display_df.iloc[:, -1] < '0']
  if len(slack_df) > 0:
    strategy_df = query_strategies(cache_clear=True)
    name_map = strategy_df[['trader', 'strategy_name']]
    slack_df = slack_df.merge(
        name_map, left_on='strat_name', right_on='strategy_name', how='left')
    market_types = sorted(slack_df['market_type'].drop_duplicates().tolist())
    for market_type in market_types:
      msg = "```" + '%s balance/margin warning\n\n' % market_type
      df = slack_df[slack_df['market_type'] == market_type]
      df.drop(columns=['config', 'price', 'strategy_name', 'market_type'],
              inplace=True)
      traders = sorted(df['trader'].drop_duplicates().tolist())
      for trader in traders:
        msg += df[df['trader'] == trader].to_string() + '\n\n'
      msg += "```"
      send_to_slack(msg, slack_receiver, 'msg', mention_list=traders)


def get_exchange_rate(balance_checker, quote):
  extra_balance = pd.DataFrame(balance_checker)
  coin_list = []
  for strat_balance in extra_balance['cur_balance']:
    for currency, balance in strat_balance.items():
      coin_list.append(currency)
  coin_list += ['USD', 'USDT']
  coin_list = list(set(coin_list))
  exchange_rate = query_exchange_rates(coin_list, quote)
  exchange_rate = dict(zip(coin_list, exchange_rate))
  return exchange_rate


def drop_excluded_strats(display_df, excluded_strats=None, excluded_markets=None):
  excluded_df = pd.DataFrame(columns=display_df.columns)
  if excluded_strats:
    excluded_df = excluded_df.append(display_df[display_df['strat_name'].isin(excluded_strats)])
    display_df = display_df[~display_df['strat_name'].isin(excluded_strats)]
  if excluded_markets:
    excluded_boolean = display_df.set_index(['market_type', 'exchange']).index.isin(excluded_markets)
    excluded_df = excluded_df.append(display_df[excluded_boolean])
    display_df = display_df[~excluded_boolean]
  return display_df, excluded_df


def get_balance_checker_loaded(strat_list, kafka_config_filename):
  kafka_config = KafkaConfig.from_cmd_config(kafka_config_filename)
  topics = get_topic_from_strat(strat_list, kafka_config)
  handler = BalanceChecker(strat_list)
  end_ts = get_timestamp()
  offset_nanosec = int(datetime.timedelta(hours=0.5).total_seconds() * 10**9)
  start_ts = end_ts - offset_nanosec
  run_from_kafka_archive(topics=topics,
                         kafka_config=kafka_config,
                         timestamp_from=start_ts,
                         timestamp_to=end_ts,
                         callback=handler.handle)
  return handler


def calculate_extra_balance(strat_list, kafka_config_filename):
  handler = get_balance_checker_loaded(strat_list, kafka_config_filename)
  balance_checker_summary = handler.calculate_all_extra_balance()
  return balance_checker_summary


def print_abnormal_strategy(all_strat_list, display_strat_list):
  strats = set(all_strat_list).difference(set(display_strat_list))
  strats = list(strats)
  strats.sort()
  print("\n\n")
  print("Abnormal Strategy:\n")
  print("\n".join(strats))


def _unfold_checker_calculation(balance_checker_summary):
  for summary in balance_checker_summary:
    if summary.get('market_type') == 'Futures':
      summary['config_curr'] = {}
      summary['price_curr'] = {}
      for currency in summary.get('min_balance').keys():
        config_curr = {}
        price_curr = {}
        for symbol in summary.get('config').keys():
          if 'portfolio' in summary.get('owner') and summary.get('exchange') == 'Binance':
            accounting_currency = 'USD'
          else:
            try:
              accounting_currency = get_accounting_currency(summary.get('market_type'),
                                                            summary.get('exchange'), symbol)
            except RuntimeError as e:
              logging.error(e)
              continue
          if accounting_currency == currency:
            config_curr.update({symbol: summary.get('config')[symbol]})
            price_curr.update({symbol: summary.get('symbol_price').get(symbol)})
        summary['config_curr'].update({currency: config_curr})
        summary['price_curr'].update({currency: price_curr})
  return balance_checker_summary


def main(argv):
  pd.set_option('display.width', 200)
  pd.set_option('max_colwidth', 200)
  strat_df = query_strategies(active=True, business_units=split_string(FLAGS.business_unit))
  if len(strat_df) == 0:
    return
  if flags.FLAGS.regex is not None:
    strat_df = strat_df.loc[strat_df['strategy_name'].str.match(flags.FLAGS.regex)]
  strat_list = strat_df['strategy_name'].tolist()
  balance_checker_summary = calculate_extra_balance(strat_list,
                                                    FLAGS.kafka_config_filename)
  exchange_rate = get_exchange_rate(balance_checker_summary, 'USD')
  balance_checker_summary = _unfold_checker_calculation(balance_checker_summary)
  display_df = convert_balance_checker_summary_for_display(
      balance_checker_summary, exchange_rate)
  print_abnormal_strategy(strat_list, display_df['strat_name'].tolist())
  display_df, excluded_df = drop_excluded_strats(
      display_df, excluded_strats=EXCLUDED_STRATS, excluded_markets=EXCLUDED_MARKETS)
  key = 'check_balance'
  if FLAGS.slack_receiver:
    curr_dt = datetime.datetime.utcnow()
    last_report_dt = None
    if os.path.exists(FLAGS.last_report_dt_record):
      with open(FLAGS.last_report_dt_record, 'r') as f:
        last_report_dt = datetime.datetime.strptime(f.read(), '%Y%m%d-%H%M%S')
    if not last_report_dt or curr_dt - last_report_dt > datetime.timedelta(minutes=FLAGS.info_alarm_frequency):
      send_balance_checker_summary(display_df, FLAGS.slack_receiver)
      with open(FLAGS.last_report_dt_record, 'w') as f:
        f.write(curr_dt.strftime('%Y%m%d-%H%M%S'))
  print('\n' * 2)
  print_balance_checker_summary(display_df[display_df['market_type'] == 'Spot'])
  print('\n' * 2)
  print_balance_checker_summary(display_df[display_df['market_type'] == 'Futures'])
  print('\n' * 2)
  print('EXCLUDED BALANCE SUMMARY')
  print_balance_checker_summary(excluded_df)



if __name__ == '__main__':
  logging.basicConfig(
      level='DEBUG',
      format='%(levelname)8s %(asctime)s %(name)s %(filename)s:%(lineno)d] %(message)s')
  flags.DEFINE_string('kafka_config_filename',
                      '../../coin_deploy/support_monitor/config/kafka_aws_config.json',
                      'kafka config')
  flags.DEFINE_string('business_unit', None, 'filter by business_unit')
  flags.DEFINE_string('slack_receiver', None, 'send_to_slack')
  flags.DEFINE_string('regex', None, '')
  flags.DEFINE_string('memcached_config',
                      "../../coin_deploy/support_accounting/db_config/memcached_config_prod.json",
                      'path of database config.')
  flags.DEFINE_float('info_alarm_frequency', 360, 'minute')
  flags.DEFINE_string('last_report_dt_record',
                      'last_report_record.json',
                      'record file path')
  app.run(main)
